package com.lagou.rpc.register.handler.impl;

import com.lagou.rpc.common.ConfigKeeper;
import com.lagou.rpc.common.listener.NodeChangeListener;
import com.lagou.rpc.common.registry.RpcRegistryHandler;
import org.apache.commons.collections4.CollectionUtils;
import org.apache.commons.collections4.MapUtils;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.recipes.cache.PathChildrenCache;
import org.apache.curator.framework.state.ConnectionState;
import org.apache.curator.retry.RetryNTimes;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.KeeperException;
import org.checkerframework.checker.units.qual.A;

import java.util.ArrayList;
import java.util.Collection;
import java.util.List;
import java.util.concurrent.*;

/**
 * 基于zookeeper实现注册中心
 */
public class ZookeeperRegistryHandler implements RpcRegistryHandler {

    private static final String LG_RPC_ZK_ROOT = "/la-rpc/";
    private static final String ZK_PATH_SPILT = "/";
    private static final List<NodeChangeListener> listenerList = new ArrayList<>();
    private String url;
    private String charset = "UTF-8";
    private CuratorFramework client;
    private volatile boolean closed;
    private List<String> serviceList = new CopyOnWriteArrayList<>();
    private static final ScheduledExecutorService REPORT_WORKER = Executors.newScheduledThreadPool(1);


    /**
     * 初始化Curator
     * @param url
     */
    public ZookeeperRegistryHandler(String url) {
        int timeout = 5000;
        CuratorFrameworkFactory.Builder builder = CuratorFrameworkFactory.builder()
                .connectString(url)
                .retryPolicy(new RetryNTimes(1, 1000))
                .connectionTimeoutMs(timeout)
                .sessionTimeoutMs(timeout);

        client = builder.build();
        client.getConnectionStateListenable().addListener((curatorFramework, connectionState) ->{
            if (ConnectionState.CONNECTED.equals(connectionState)) {
                System.out.println("注册中心连接成功");
            }
        });
        //启动
        client.start();
        this.url = url;

        //定时上报
//        ConfigKeeper instance = ConfigKeeper.getInstance();
//        boolean consumerSide = instance.isConsumerSide();
//        int interval = instance.getInterval();
        //只有消费者端才上报耗时统计
//        if (consumerSide && interval > 0) {
//            REPORT_WORKER.scheduleWithFixedDelay(() -> {
//                ConcurrentHashMap<String, Long> metricMap = RequestMetrics.getInstance().getMetricMap();
//                if (MapUtils.isEmpty(metricMap)) {
//                    return;
//                }
//                System.out.println("自动上报节点耗时日志:" + metricMap);
//                metricMap.entrySet().forEach(entry -> {
//                    String address = entry.getKey();
//                    Long cost = entry.getValue();
//                    // 先创建路径
//                    String zkPath = metricsPath();
//                    if (!exists(zkPath)) {
//                        create(zkPath, false);
//                    }
//                    String instancePath = zkPath + ZK_PATH_SPILT + address;
//
//                    updateWithData(instancePath, cost.toString(), true);
//                });
//            }, interval, interval, TimeUnit.SECONDS);
//        }
    }



    @Override
    public boolean registry(String service, String ip, int port) {
        String zkPath = providePath(service);
        if (!exists(zkPath)) {
            create(zkPath, false);
        }
        String instancePath = zkPath + ZK_PATH_SPILT + ip + ":" + port;
        create(instancePath, true);
        return true;
    }

    private void create(String zkPath, boolean ephemeral) {
        if (ephemeral) {
            this.createEphemeral(zkPath);
        } else {
            this.createPersistent(zkPath);
        }
    }

    private void createEphemeral(String zkPath) {
        try {
            client.create().creatingParentsIfNeeded().withMode(CreateMode.EPHEMERAL).forPath(zkPath);
        } catch (KeeperException.NodeExistsException e) {
            System.out.println("路径[" + zkPath + "]已存在");
        } catch (Exception e) {
            throw new IllegalStateException(e.getMessage(), e);
        }
    }

    public void createPersistent(String path) {
        try {
            client.create().creatingParentsIfNeeded().withMode(CreateMode.PERSISTENT).forPath(path);
        } catch (KeeperException.NodeExistsException e) {
            System.out.println("路径[" + path + "]已存在");
        } catch (Exception e) {
            throw new IllegalStateException(e.getMessage(), e);
        }
    }

    public boolean isConnected() {
        return this.client.getZookeeperClient().isConnected();
    }

    private String providePath(String service) {
        return LG_RPC_ZK_ROOT + service + ZK_PATH_SPILT + "provider";
    }

    public String getPath() {
        return url;
    }
    public boolean exists(String path) {
        try {
            if (client.checkExists().forPath(path) != null) {
                return true;
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
        return false;
    }


    @Override
    public List<String> discovery(String service) {
        /**
         * 完成服务地址的查找
         * /lg-rpc/com.lagou.rpc.api.UserService/provider
         */
        String path = providePath(service);
        try {
            //第一次为空，需要从zk获取，后续通过watcher机制更新
            if (CollectionUtils.isEmpty(serviceList)) {
                System.out.println("首次从注册中心查找服务地址...");
                serviceList = client.getChildren().forPath(path);
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
        this.registryWatch(service, path);
        return serviceList;

    }

    private void registryWatch(String service, String path) {
        PathChildrenCache nodeCache = new PathChildrenCache(client, path, true);
        nodeCache.getListenable().addListener((client, pathChildrenCacheEvent)->{
            //更新本地缓存
            serviceList = client.getChildren().forPath(path);
            listenerList.stream().forEach(nodeChangeListener -> {
                System.out.println("节点变化，开始通知业务");
                nodeChangeListener.notify(service, serviceList, pathChildrenCacheEvent);
            });
        });
        // /lg-edu-rpc/com.lagou.edu.api.UserService/provider/127.0.0.1:8990
        /*
         * StartMode：初始化方式
         * POST_INITIALIZED_EVENT：异步初始化。初始化后会触发事件
         * NORMAL：异步初始化
         * BUILD_INITIAL_CACHE：同步初始化
         * */
        try {
            nodeCache.start(PathChildrenCache.StartMode.BUILD_INITIAL_CACHE);
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    @Override
    public void addListener(NodeChangeListener listener) {
        listenerList.add(listener);
    }

    @Override
    public void destroy() {
        if (null != client) {
            client.close();
        }
    }

    public void close() {
        if (closed) {
            return;
        }
        this.closed = true;
        this.client.close();
    }

}
